package com.xxl.job.admin.core.scheduler;

import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
import com.xxl.job.admin.core.thread.*;
import com.xxl.job.admin.core.util.I18nUtil;
import com.xxl.job.core.biz.ExecutorBiz;
import com.xxl.job.core.biz.client.ExecutorBizClient;
import com.xxl.job.core.enums.ExecutorBlockStrategyEnum;
import com.xxl.job.core.handler.IJobHandler;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * @author xuxueli 2018-10-28 00:18:17
 */

public class XxlJobScheduler  {
    private static final Logger logger = LoggerFactory.getLogger(XxlJobScheduler.class);

    /**
     * 点该方法，就知道该方法是在什么时候执行了
     * 其实，写在：实现 InitializingBean 的 afterPropertiesSet()方法中调用，还是不理解为什么放在这个方法里面
     */
    public void init() throws Exception {
        // init i18n
        // 国际化
        // 配置参数： xxl.job.i18n=zh_CN, 这里设置为中文简体
        initI18n();

        // admin trigger pool start  admin 触发线程池启动,仅仅是启动
        // 1.初始化了：fastTriggerPool，slowTriggerPool 两个线程池
        // 2.这两个的使用，可以再下面 失败任务处理的守护线程中看到
        // 3.作用是，当失败任务重试次数超过10，就会有fastTriggerPool 转为 slowTriggerPool
        //      最终调用在: JobTriggerPoolHelper类中的addTrigger方法调用
        // 4.两个线程池的区别：
        //     4.1 fastTriggerPool:
        //              最大线程数：200(默认)
        //              阻塞队列长度：1000
        //     4.2 slowTriggerPool:
        //              最大线程数：100(默认)
        //              阻塞队列长度：2000
        JobTriggerPoolHelper.toStart();

        // admin registry monitor run
        // start()里面有：
        // 1. 一个注册/删除线程池: registryOrRemoveThreadPool  该线程池只是给实例化了
        // 2. 一个registry的守护线程: registryMonitorThread
        //   2.1 30s 执行一次
        //   2.2 主要的目的是：检查group表和registry表中 执行器实例列表是否处于在线状态以及离线剔除（删除无效的地址，更新最新的地址）
        JobRegistryHelper.getInstance().start();

        // admin fail-monitor run
        // 失败任务处理的守护线程：
        // 1. 10s 执行一次;
        // 2. 通过去 xxl_job_log 表中捞取处理是失败的任务
        // 3. 通过路由策略，发送run()请求;
        // 4. 之后 因为执行器和admin是通过netty通信的;
        //    执行器接到请求之后，就会在
        //          EmbedHttpServerHandler extends SimpleChannelInboundHandler
        //    的channelRead0()方法中处理接收请求;并利用 writeResponse()方法 将处理返回给调用房
        // 5. 处理请求就是找对对应的 JobHandler, JobHandler在执行器实例启动的时候已经封装再也一个map了
        //       这个 map 在 XxlJobExecutor类中： jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>();
        // 6. 然后利用 JDK动态代理 invoke对应的任务方法
        JobFailMonitorHelper.getInstance().start();

        // admin lose-monitor run ( depend on JobTriggerPoolHelper ) 任务回调和任务丢失标记
        // 1. 会开辟一个 callbackThreadPool 回调线程池
        // 2. 会启动一个守护线程: 用于 任务结果丢失处理：调度记录停留在 "运行中" 状态超过10min，且对应执行器心跳注册失败不在线，则将本地调度主动标记失败；
        // 3. 需要等待50s, 先让 JobTriggerPoolHelper.toStart() 中的两个线程池开辟完成
        // 4. 60s执行一次
        JobCompleteHelper.getInstance().start();

        // admin log report start  任务执行报表
        // 主要的作用就是生成一个 日志报表 xxl_job_log_report
        // 1. 更新 统计近3天每天的执行日志 xxl_job_log中 成功,失败,运行中的日志个数
        // 2. 自动删除 超过30天的日志信息
        // 3. 每毫秒执行一次; 我自己的理解是，可能有些任务就是每秒，每毫秒执行一次，如果查看报表时候没有更新，多少会有误差
        //      不过本身就不需要太准确的值
        JobLogReportHelper.getInstance().start();

        // start-schedule  ( depend on JobTriggerPoolHelper )
        // 这里就是 admin 发起调度执行 注册任务(job_info表) 的地方了。
        // 任务调度器
        // 工作流程：
        //      1. 周期性的遍历所有的jobInfo这个表，通过数据库的行锁和事务一致性，通过for update 来保证多个调度中心集群在同一时间内只有一个调度中心在调度任务；
        //      2. 周期性的遍历所有的jobInfo这个表，读取触发时间小于nowtime+5s这个时间之前的所有任务，
        //            预读数据，从数据库中读取当前截止到五秒后内会执行的job信息，并且读取分页大小为preReadCount=6000条数据
        //      3. 然后进行引入以下触发机制判断：
        //         将当前时间与下次调度时间对比，有如下三种情况
        //         三种触发任务机制：
        //            1)nowtime-TriggerNextTime()>PRE_READ_MS(5s) 既超过有效误差内
        //                  - 1.1) 则查看当前任务的失效调度策略，若为立即重试一次，则立即触发调度任务，且触发类型为misfire;若为DO_NOTHING=过期啥也不干
        //                  - 2.2) 刷新上一次触发 和 下一次待触发时间
        //            2)nowtime-TriggerNextTime()<PRE_READ_MS(5s) 既没有超过有效误差，
        //                  - 1.1) 立即调度，触发执行器任务;
        //                  - 1.2) 刷新上一次触发 和 下一次待触发时间;
        //                  - 1.3) 如果下一次触发在五秒内，直接放进时间轮里面待调度;
        //                           -- ① 求当前任务下一次触发时间所处一分钟的第N秒;
        //                           -- ② 将当前任务ID和ringSecond放进时间轮里面;
        //                           -- ③ 刷新上一次触发 和 下一次待触发时间.
        //            3)nowtime<TriggerNextTime() : (当前时间 小于 下一次触发时间) 则说明这个任务马上就要触发了
        //                  - 1.1) 求当前任务下一次触发时间所处一分钟的第N秒;
        //                  - 1.2) 将当前任务id和ringSecond放进 时间轮 里面;
        //                  - 1.3) 刷新上一次触发 和 下一次待触发时间
        //                            随后将快要触发的任务放到时间轮上，
        //                            时间轮由key(将要触发的时间s)，value(在当前触发s的所有任务id集合)，
        //                            然后更新这个任务的下一次触发时间;
        //      4. 更新数据库执行器信息，如trigger_last_time、trigger_next_time
        //      5. 提交数据库事务，释放数据库select for update 排它锁
        //
        // ringThread 线程：根据时间轮执行job（取数据执行）
        //     首先时间轮数据格式为：Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>()
        //       key:  将要触发的时间s
        //       value:在当前触发s的所有任务id集合
        //   执行步骤:
        //      1. 获取当前所处的一分钟第几秒，然后for两次，第二次是为了重跑前面一个刻度没有被执行的的job list，避免前面的刻度遗漏了;
        //      2. 触发执行器
        //      3. 清除当前刻度列表的数据
        JobScheduleHelper.getInstance().start();

        logger.info(">>>>>>>>> init xxl-job admin success.");
    }

    /**
     * 为了在销毁时，释放资源：
     *    主要是为了释放线程资源，优雅的释放
     */
    public void destroy() throws Exception {

        // stop-schedule
        JobScheduleHelper.getInstance().toStop();

        // admin log report stop
        JobLogReportHelper.getInstance().toStop();

        // admin lose-monitor stop
        JobCompleteHelper.getInstance().toStop();

        // admin fail-monitor stop
        JobFailMonitorHelper.getInstance().toStop();

        // admin registry stop
        JobRegistryHelper.getInstance().toStop();

        // admin trigger pool stop
        JobTriggerPoolHelper.toStop();

    }

    // ---------------------- I18n ----------------------
    // 国际化 阻塞处理策略
    private void initI18n(){
        for (ExecutorBlockStrategyEnum item:ExecutorBlockStrategyEnum.values()) {
            item.setTitle(I18nUtil.getString("jobconf_block_".concat(item.name())));
        }
    }

    // ---------------------- executor-client ----------------------
    private static ConcurrentMap<String, ExecutorBiz> executorBizRepository = new ConcurrentHashMap<String, ExecutorBiz>();
    public static ExecutorBiz getExecutorBiz(String address) throws Exception {
        // valid
        if (address==null || address.trim().length()==0) {
            return null;
        }

        // load-cache
        address = address.trim();
        ExecutorBiz executorBiz = executorBizRepository.get(address);
        if (executorBiz != null) {
            return executorBiz;
        }

        // set-cache
        executorBiz = new ExecutorBizClient(address, XxlJobAdminConfig.getAdminConfig().getAccessToken());

        executorBizRepository.put(address, executorBiz);
        return executorBiz;
    }

}
